package com.bw.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.app.functions.DimAsyncFunction;
import com.bw.bean.OrderWide;
import com.bw.bean.ProductStats;
import com.bw.bean.PaymentWide;
import com.bw.bean.ProductStats;
import com.bw.utils.ClickHouseUtil;
import com.bw.utils.DateTimeUtil;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.text.ParseException;
import java.time.Duration;
import java.util.Date;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;

public class ProductStatsApp {
    public static void main(String[] args) throws Exception {

        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //        开启CK，创建文件夹
//        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:9820/gmall-flink/ck"));
//         每隔5秒保存一次
//        env.enableCheckpointing(5000L);
//        精准消费一次
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        配置超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(100000L);
//        同时最大可以运行几个ck
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
//        上一个ck的尾部到下一个ck的头之间间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
        // env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 2. 读取kafka 多条流
        String groupId = "product_stats_app_2105b2";
        String pageViewSourceTopic = "dwd_page_log";
        String favorInfoSourceTopic = "dwd_favor_info";
        String cartInfoSourceTopic = "dwd_cart_info";
        String ProductStatsSourceTopic = "dwm_order_wide";
        String paymentWideSourceTopic = "dwm_payment_wide";
        String refundInfoSourceTopic = "dwd_order_refund_info";
        String commentInfoSourceTopic = "dwd_comment_info";

        DataStreamSource<String> pvDS = env.addSource(MyKafkaUtil.getKafkaConsumer(pageViewSourceTopic, groupId));
        DataStreamSource<String> favorDS = env.addSource(MyKafkaUtil.getKafkaConsumer(favorInfoSourceTopic, groupId));
        DataStreamSource<String> cartDS = env.addSource(MyKafkaUtil.getKafkaConsumer(cartInfoSourceTopic, groupId));
        DataStreamSource<String> ProductStatsDS = env.addSource(MyKafkaUtil.getKafkaConsumer(ProductStatsSourceTopic, groupId));
        DataStreamSource<String> paymentWideDS = env.addSource(MyKafkaUtil.getKafkaConsumer(paymentWideSourceTopic, groupId));
        DataStreamSource<String> refundDS = env.addSource(MyKafkaUtil.getKafkaConsumer(refundInfoSourceTopic, groupId));
        DataStreamSource<String> commentDS = env.addSource(MyKafkaUtil.getKafkaConsumer(commentInfoSourceTopic, groupId));

        // 3. 把每条流转成相同的JavaBean对象  什么点击，要判断item_type和page_id。曝光要输出多条
        SingleOutputStreamOperator<ProductStats> pvToProDS = pvDS.flatMap(new FlatMapFunction<String, ProductStats>() {
            @Override
            public void flatMap(String value, Collector<ProductStats> collector) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                // 什么是点击----> page_id=detail && item_type=sku_id
                JSONObject page = jsonObject.getJSONObject("page");
                String itemType = page.getString("item_type");
                String pageId = page.getString("page_id");
                Long ts = jsonObject.getLong("ts");
                if ("good_detail".equals(pageId) && "sku_id".equals(itemType)) {
                    Long sku_id = page.getLong("item");
                    collector.collect(
                            ProductStats.builder()
                                    .sku_id(sku_id)
                                    .click_ct(1L)
                                    .ts(ts)
                                    .build());
                }
                // 取出曝光
                JSONArray displays = jsonObject.getJSONArray("displays");
                if (displays != null && displays.size() > 0) {
                    for (int i = 0; i < displays.size(); i++) {
                        JSONObject display = displays.getJSONObject(i);
                        // 只要商品曝光，不要活动曝光
                        if ("sku_id".equals(display.getString("item_type"))) {
                                Long sku_id = display.getLong("item");
                                collector.collect(ProductStats
                                        .builder()
                                        .sku_id(sku_id)
                                        .display_ct(1L)
                                        .ts(ts)
                                        .build());


                        }
                    }
                }

            }
        });
        // 处理收藏
        SingleOutputStreamOperator<ProductStats> favorToProDS = favorDS.map(new MapFunction<String, ProductStats>() {
            @Override
            public ProductStats map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                return ProductStats.builder()
                        .sku_id(jsonObject.getLong("sku_id"))
                        .favor_ct(1L)
                        .ts(DateTimeUtil.toTs(jsonObject.getString("create_time")))
                        .build();
            }
        });

        // 处理购物车
        SingleOutputStreamOperator<ProductStats> cratToProDS = cartDS.map(new MapFunction<String, ProductStats>() {
            @Override
            public ProductStats map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                return ProductStats.builder()
                        .sku_id(jsonObject.getLong("sku_id"))
                        .cart_ct(1L)
                        .ts(DateTimeUtil.toTs(jsonObject.getString("create_time")))
                        .build();
            }
        });
        // 订单宽表流
        SingleOutputStreamOperator<ProductStats> ProductStatsToProDS = ProductStatsDS.map(new MapFunction<String, ProductStats>() {
            @Override
            public ProductStats map(String value) throws Exception {
                OrderWide orderWide = JSON.parseObject(value, OrderWide.class);
                Set<Long> orderIds = new HashSet<>();
                orderIds.add(orderWide.getOrder_id());
                return ProductStats.builder()
                        .sku_id(orderWide.getSku_id())
                        .order_sku_num(orderWide.getSku_num())
                        .orderIdSet(orderIds)
                        .order_amount(orderWide.getSplit_total_amount())
                        .ts(DateTimeUtil.toTs(orderWide.getCreate_time()))
                        .build();
            }
        });
        // 支付流
        SingleOutputStreamOperator<ProductStats> paymentWideToProDS = paymentWideDS.map(new MapFunction<String, ProductStats>() {
            @Override
            public ProductStats map(String value) throws Exception {
                PaymentWide paymentWide = JSON.parseObject(value, PaymentWide.class);
                Set<Long> orderIds = new HashSet<>();
                orderIds.add(paymentWide.getOrder_id());
                return ProductStats.builder()
                        .sku_id(paymentWide.getSku_id())
                        .payment_amount(paymentWide.getSplit_total_amount())
                        .paidOrderIdSet(orderIds)
                        .ts(DateTimeUtil.toTs(paymentWide.getPayment_create_time()))
                        .build();
            }
        });
        // 退款
        SingleOutputStreamOperator<ProductStats> refundToProDS = refundDS.map(new MapFunction<String, ProductStats>() {
            @Override
            public ProductStats map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                Long skuId = jsonObject.getLong("sku_id");
                BigDecimal totalAmount = jsonObject.getBigDecimal("refund_amount");
                Long orderId = jsonObject.getLong("order_id");
                Set<Long> orderIds = new HashSet<>();
                orderIds.add(orderId);
                return ProductStats.builder()
                        .sku_id(skuId)
                        .refund_amount(totalAmount)
                        .refundOrderIdSet(orderIds)
                        .ts(DateTimeUtil.toTs(jsonObject.getString("create_time")))
                        .build();
            }
        });
        SingleOutputStreamOperator<ProductStats> commentToProDS = commentDS.map(new MapFunction<String, ProductStats>() {
            @Override
            public ProductStats map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                Long skuId = jsonObject.getLong("sku_id");
                Long orderId = jsonObject.getLong("order_id");
                String appraise = jsonObject.getString("appraise");
                Set<Long> orderIds = new HashSet<>();
                orderIds.add(orderId);
                Long goodCt = 0L;
                if ("1204".equals(appraise)) {
                    goodCt = 1L;
                }
                return ProductStats.builder()
                        .sku_id(skuId)
                        .comment_ct(1L)
                        .good_comment_ct(goodCt)
                        .ts(DateTimeUtil.toTs(jsonObject.getString("create_time")))
                        .build();
            }
        });

        // 4. 把所有流union在一起
        DataStream<ProductStats> unionDS = pvToProDS.union(favorToProDS, cratToProDS, ProductStatsToProDS, paymentWideToProDS, refundToProDS, commentToProDS);

        // 5. 添加水印
        SingleOutputStreamOperator<ProductStats> unionWithWmDS = unionDS.assignTimestampsAndWatermarks(WatermarkStrategy.<ProductStats>forBoundedOutOfOrderness(Duration.ofSeconds(1)).withTimestampAssigner(new SerializableTimestampAssigner<ProductStats>() {
            @Override
            public long extractTimestamp(ProductStats productStats, long l) {
                return productStats.getTs();
            }
        }));
        // 6. Keyby 开窗 聚合，去重
        SingleOutputStreamOperator<ProductStats> reduceWindowDS = unionWithWmDS.keyBy(new KeySelector<ProductStats, Long>() {
            @Override
            public Long getKey(ProductStats productStats) throws Exception {
                return productStats.getSku_id();
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10))).reduce(new ReduceFunction<ProductStats>() {

            @Override
            public ProductStats reduce(ProductStats stats1, ProductStats stats2) throws Exception {
                stats1.setDisplay_ct(stats1.getDisplay_ct() + stats2.getDisplay_ct());
                stats1.setClick_ct(stats1.getClick_ct() + stats2.getClick_ct());
                stats1.setCart_ct(stats1.getCart_ct() + stats2.getCart_ct());
                stats1.setFavor_ct(stats1.getFavor_ct() + stats2.getFavor_ct());
                stats1.setOrder_amount(stats1.getOrder_amount().add(stats2.getOrder_amount()));
                stats1.getOrderIdSet().addAll(stats2.getOrderIdSet());
                //stats1.setOrder_ct(stats1.getOrderIdSet().size() + 0L);
                stats1.setOrder_sku_num(stats1.getOrder_sku_num() + stats2.getOrder_sku_num());
                stats1.setPayment_amount(stats1.getPayment_amount().add(stats2.getPayment_amount()));

                stats1.getRefundOrderIdSet().addAll(stats2.getRefundOrderIdSet());
                //stats1.setRefund_order_ct(stats1.getRefundOrderIdSet().size() + 0L);
                stats1.setRefund_amount(stats1.getRefund_amount().add(stats2.getRefund_amount()));

                stats1.getPaidOrderIdSet().addAll(stats2.getPaidOrderIdSet());
                //stats1.setPaid_order_ct(stats1.getPaidOrderIdSet().size() + 0L);

                stats1.setComment_ct(stats1.getComment_ct() + stats2.getComment_ct());
                stats1.setGood_comment_ct(stats1.getGood_comment_ct() + stats2.getGood_comment_ct());
                return stats1;
            }
        }, new WindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {
            @Override
            public void apply(Long aLong, TimeWindow timeWindow, Iterable<ProductStats> iterable, Collector<ProductStats> collector) throws Exception {
                String start = DateTimeUtil.toYMDhms(new Date(timeWindow.getStart()));
                String end = DateTimeUtil.toYMDhms(new Date(timeWindow.getEnd()));
                ProductStats productStats = iterable.iterator().next();
                productStats.setStt(start);
                productStats.setEdt(end);
                // 设置去重订单数
                productStats.setOrder_ct((long) productStats.getOrderIdSet().size());
                productStats.setPaid_order_ct((long) productStats.getPaidOrderIdSet().size());
                productStats.setRefund_order_ct((long) productStats.getRefundOrderIdSet().size());

                collector.collect(productStats);
            }
        });
        // 7. 关联维度
        // 关联SKU
        SingleOutputStreamOperator<ProductStats> productStatsWithSkuDS = AsyncDataStream.unorderedWait(
                reduceWindowDS, new DimAsyncFunction<ProductStats>("DIM_SKU_INFO") {
                    @Override
                    public void join(ProductStats productstats, JSONObject dimInfo) throws ParseException {
                        productstats.setSku_name(dimInfo.getString("SKU_NAME"));
                        productstats.setSku_price(dimInfo.getBigDecimal("PRICE"));
                        productstats.setSpu_id(dimInfo.getLong("SPU_ID"));
                        productstats.setTm_id(dimInfo.getLong("TM_ID"));
                        productstats.setCategory3_id(dimInfo.getLong("CATEGORY3_ID"));
                    }

                    @Override
                    public String getKey(ProductStats productstats) {
                        return String.valueOf(productstats.getSku_id());
                    }
                }, 120, TimeUnit.SECONDS);
        //
        //5.4 关联SPU维度
        SingleOutputStreamOperator<ProductStats> productStatsWithSpuDS = AsyncDataStream.unorderedWait(
                productStatsWithSkuDS, new DimAsyncFunction<ProductStats>("DIM_SPU_INFO") {
                    @Override
                    public void join(ProductStats productstats, JSONObject jsonObject) throws ParseException {
                        productstats.setSpu_name(jsonObject.getString("SPU_NAME"));
                    }

                    @Override
                    public String getKey(ProductStats productstats) {
                        return String.valueOf(productstats.getSpu_id());
                    }
                }, 120, TimeUnit.SECONDS);
//
//        //5.5 关联TM维度
        SingleOutputStreamOperator<ProductStats> productStatsWithTmDS = AsyncDataStream.unorderedWait(
                productStatsWithSpuDS, new DimAsyncFunction<ProductStats>("DIM_BASE_TRADEMARK") {
                    @Override
                    public void join(ProductStats productstats, JSONObject jsonObject) throws ParseException {
                        productstats.setTm_name(jsonObject.getString("TM_NAME"));
                    }

                    @Override
                    public String getKey(ProductStats productstats) {
                        return String.valueOf(productstats.getTm_id());
                    }
                }, 60, TimeUnit.SECONDS);
//
//        //5.6 关联Category维度
        SingleOutputStreamOperator<ProductStats> productStatsWithCategory3DS = AsyncDataStream.unorderedWait(
                productStatsWithTmDS, new DimAsyncFunction<ProductStats>("DIM_BASE_CATEGORY3") {
                    @Override
                    public void join(ProductStats productstats, JSONObject jsonObject) throws ParseException {
                        productstats.setCategory3_name(jsonObject.getString("NAME"));

                    }

                    @Override
                    public String getKey(ProductStats productstats) {
                        return String.valueOf(productstats.getCategory3_id());
                    }
                }, 120, TimeUnit.SECONDS);

        // 8、写入ClickHouse
        productStatsWithCategory3DS.print();
        productStatsWithCategory3DS.addSink(ClickHouseUtil.sink("insert into table product_stats_2105b values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));
        // 9. 启动任务
        env.execute("ProductStatsApp");
    }
}
